package hbec.kettle.etl.engine.schedule;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import hbec.kettle.etl.common.constants.DataBase;
import hbec.kettle.etl.common.exception.ExceptionTracker;
import hbec.kettle.etl.common.plugin.Result;
import hbec.kettle.etl.engine.conf.JobEntityConf;
import hbec.kettle.etl.engine.conf.TransEntityConf;
import hbec.kettle.etl.engine.conf.WriterConf;
import hbec.kettle.etl.engine.plugin.BufferedLineExchanger;
import hbec.kettle.etl.engine.plugin.Line;
import hbec.kettle.etl.plugins.common.DALUtils;
import hbec.kettle.etl.plugins.writer.mysqlwriter.MysqlWriter;

public class WriterWorker extends PluginWorker implements Runnable {
	private static final Logger logger = LoggerFactory.getLogger(WriterWorker.class);
	
	private JobEntityConf job;
	Map<DataBase, String> watchSumTablesOfDb;
	private DataBase db;
	private String cpTable;
	private Line line;
	
	public WriterWorker(WriterConf writerConf, Line line) {
		this.job = writerConf.getJob();
		this.watchSumTablesOfDb = writerConf.getWatchSumTablesOfDb();
		for (Map.Entry<DataBase, String> e : writerConf.getCheckPointTableOfDb().entrySet()) {
			this.db = e.getKey();
			this.cpTable = e.getValue();
		}
		this.line = line;
	}
	
	@Override
	public void run() {
		int transCount = job.getTransEntityConfs().size();
		
		CountDownLatch latch = new CountDownLatch(transCount);	
		ExecutorService threadPool = Executors.newFixedThreadPool(transCount);
		
		List<Future<Result>> res = new ArrayList<>();
		for (TransEntityConf transEntityConf : job.getTransEntityConfs()) {
			Future<Result> future = threadPool.submit(new MysqlWriter(watchSumTablesOfDb, transEntityConf, line, latch));
			res.add(future);
		}
		threadPool.shutdown();
		
		boolean isFinish = false;
		try {
			isFinish = latch.await(5L, TimeUnit.MINUTES);
		} catch (InterruptedException e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			if (isFinish && res != null) {
				boolean ret = false;
				for (Future<Result> r : res) {
					try {
						ret = r.get().result;
						if (!ret) 
							break;
					} catch (InterruptedException e) {
						logger.error(ExceptionTracker.trace(e));
					} catch (ExecutionException e) {
						logger.error(ExceptionTracker.trace(e));
					}
				}
				
				if (ret) {
					int i = DALUtils.setCheckPoint(db, cpTable, line.getObject().split("\\.")[1].trim(), line.getMaxPosition());
					if (i > 0) {
						//LineExchanger.popLine(line);
						Engine.workingJob.remove(line.getObject());
						BufferedLineExchanger.removeLineBuffer(line.getObject(), line);
						BufferedLineExchanger.removeFailedObject(line.hashCode());
						logger.info("checkpoint save success.");
					}
					else 
						logger.info("checkpoint save fail");
				}
			} else {
				BufferedLineExchanger.addFailedObject(line.getObject(), line);
			}
		}
	}

}
